package com.hualife;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import org.apache.commons.lang.StringUtils;
import org.apache.flume.Context;
import org.apache.flume.Event;
import org.apache.flume.interceptor.Interceptor;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;


import java.util.ArrayList;
import java.util.List;
import java.util.Map;

public class CustomInterceptor implements Interceptor{
    private static final Logger logger = LoggerFactory.getLogger(CustomInterceptor.class);
    // 声明一个存放事件的集合
    private List<Event> addHeaderEvents;

    /**
     * 自定义参数
     */
    //private String p1;
    //private String p2;

    @Override
    public void initialize() {
        addHeaderEvents = new ArrayList<Event>();
    }

    // 单个事件拦截
    @Override
    public Event intercept(Event event) {
        // 1.获取事件中的头信息
        Map<String, String> headers = event.getHeaders();
        for (String header:headers.keySet()){
            logger.info("事件头信息的key:{},value:{}",header,headers.get(header));
        }
        //logger.info(headers.keySet().toString());
        // 2.获取事件中的body信息
        String body = new String(event.getBody());
        logger.info("事件body信息:{}", body);
        //设置赋值自定义配置
        String branch_type = null;

        //把字符串转化为JSON格式的数据结构（key-value 结构）
        try {
            JSONObject jb = JSONObject.parseObject(body);
            //SimpleDateFormat format = new SimpleDateFormat( "yyyyMMdd" );
            branch_type = jb.getString( "appid" );
            //createTime = format.format( jb.getLong( p2 ) );
        } catch (Exception e) {
            logger.error( "JSON解析异常,Message:{}",e.getMessage(),e );
            System.out.println("迭代队列————>JSON解析异常————>Message:"+ JSON.toJSONString(e));
        }

        if (StringUtils.isBlank( branch_type )) {
            branch_type = "unknown";
        }
        headers.put( "branch_type",branch_type );

        event.setHeaders(headers);
        Map<String, String> headers_new = event.getHeaders();
        for (String header:headers_new.keySet()){
            logger.info("新的事件头信息的key:{},value:{}",header,headers_new.get(header));
        }
        return event;
    }

    // 批量事件拦截
    @Override
    public List<Event> intercept(List<Event> events) {
        // 1.清空集合
        addHeaderEvents.clear();
        // 2.遍历events
        for (Event event : events) {
            // 3.为每个事件添加头信息
            addHeaderEvents.add(intercept(event));
        }
        return addHeaderEvents;
    }

    @Override
    public void close() {

    }

    public static class Builder implements Interceptor.Builder {

        @Override
        public Interceptor build() {
            return new CustomInterceptor();
        }

        @Override
        public void configure(Context context) {
        }
    }
}
